package com.amigo.online.provider.compare.quartzjob;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

import org.springframework.beans.factory.annotation.Autowired;

import com.amigo.online.provider.common.ConstantValue;
import com.amigo.online.provider.common.compare.AbstractBasicCompareService;
import com.amigo.online.provider.common.entity.UserMongoInfo;
import com.amigo.online.provider.common.entity.UserMysqlInfo;
import com.amigo.online.provider.common.entity.UserSolrInfo;
import com.amigo.online.provider.compare.service.UserComparisonService;
import com.amigo.online.provider.kafka.common.KafkaConstantValue;
import com.amigo.online.provider.kafka.model.CompareFailMessage;
import com.amigo.online.provider.sync.handle.data.mongo.service.UserMongoService;
import com.amigo.online.provider.sync.handle.data.solr.service.UserSolrService;

/**
 * @author amigo
 */

public class AbstractUserComparisonJob extends AbstractCompareJob {

    @Autowired
    ExecutorService executorService;

    @Autowired
    UserComparisonService userComparisonService;

    @Autowired
    UserMongoService userMongoService;

    @Autowired
    UserSolrService userSolrService;


    protected void compare(List<UserMongoInfo> userMongoList, List<UserSolrInfo> userSolrList,
                           List<UserMysqlInfo> userMysqlList) {
        Map<Long, UserMongoInfo> mongoMap = userMongoList.stream()
                .collect(Collectors.toMap(UserMongoInfo::getId, userMongoInfo -> userMongoInfo));
        Map<Long, UserSolrInfo> solrMap = userSolrList.stream()
                .collect(Collectors.toMap(UserSolrInfo::getId, userSolrInfo -> userSolrInfo));
        for (UserMysqlInfo temp : userMysqlList) {
            UserMongoInfo userMongoInfo = mongoMap.get(temp.getId());
            UserSolrInfo userSolrInfo = solrMap.get(temp.getId());
            Map<String, Object> solrWrongMap = new HashMap<>(32);
            Map<String, Object> mongoWrongMap = new HashMap<>(32);
            AbstractBasicCompareService.compare(temp, userMongoInfo, mongoWrongMap);
            AbstractBasicCompareService.compare(temp, userSolrInfo, solrWrongMap);

            if (userMongoInfo != null) {
                AbstractBasicCompareService.compare(temp, userMongoInfo, mongoWrongMap);
                if (!mongoWrongMap.isEmpty()) {
                    CompareFailMessage compareFailMessage = new CompareFailMessage(ConstantValue.TBL_UPDATE,
                            ConstantValue.USER_TBL, ConstantValue.SOURCE_MONGO, mongoWrongMap, temp.getId());
                    kafkaMessageSender.failLogSend(KafkaConstantValue.COMPARSION_MONGO_USER_FAIL, compareFailMessage);
                }
            } else {
                CompareFailMessage compareFailMessage = new CompareFailMessage(ConstantValue.TBL_INSERT,
                        ConstantValue.USER_TBL, ConstantValue.SOURCE_MONGO, "", temp.getId());
                kafkaMessageSender.failLogSend(KafkaConstantValue.COMPARSION_MONGO_USER_FAIL, compareFailMessage);
            }
            if (userSolrInfo != null) {
                AbstractBasicCompareService.compare(temp, userSolrInfo, solrWrongMap);
                if (!mongoWrongMap.isEmpty()) {
                    CompareFailMessage compareFailMessage = new CompareFailMessage(ConstantValue.TBL_UPDATE,
                            ConstantValue.USER_TBL, ConstantValue.SOURCE_SOLR, mongoWrongMap, temp.getId());
                    kafkaMessageSender.failLogSend(KafkaConstantValue.COMPARSION_SOLR_USER_FAIL, compareFailMessage);
                }
            } else {
                CompareFailMessage compareFailMessage = new CompareFailMessage(ConstantValue.TBL_INSERT,
                        ConstantValue.USER_TBL, ConstantValue.SOURCE_SOLR, "", temp.getId());
                kafkaMessageSender.failLogSend(KafkaConstantValue.COMPARSION_SOLR_USER_FAIL, compareFailMessage);

            }

        }
    }
}
